package com.gin.monitor.agg.connector.mysql;

import com.gin.monitor.agg.utils.GenericClassUtil;
import com.gin.monitor.agg.vo.AvgSpeedInfo;
import com.gin.monitor.agg.vo.OutOfLimitSpeedInfo;
import com.gin.monitor.agg.vo.RepetitionCarWarning;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.*;
import java.text.SimpleDateFormat;
import java.util.Date;


/**
 * @author gin
 * @date 2021/5/10
 */
public abstract class AbstractHbaseSink<T> extends RichSinkFunction<T> {

    Connection conn = null;
    PreparedStatement pst = null;
    /**
     * 注意, 需要子类实现父类才能获取到泛型 T
     */
    Class<T> tClass = GenericClassUtil.getSuperClassGenericType(this.getClass());

    @Override
    public void open(Configuration parameters) throws Exception {
        //Class<T> tClass2 = (Class<T>) ((ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[0];
        conn = DriverManager.getConnection("jdbc:mysql://node01/test", "root", "123456");
        if (tClass.getName().equals(OutOfLimitSpeedInfo.class.getName())) {
            pst = conn.prepareStatement("insert into t_speeding_info (car,monitor_id,road_id,real_speed,limit_speed,action_time) values (?,?,?,?,?,?)");
        }
        if (tClass.getName().equals(AvgSpeedInfo.class.getName())) {
            pst = conn.prepareStatement("insert into t_average_speed (start_time,end_time,monitor_id,avg_speed,car_count) values (?,?,?,?,?)");
        }
        if (tClass.getName().equals(RepetitionCarWarning.class.getName())) {
            pst = conn.prepareStatement("insert into t_violation_list (car,violation,create_time) values (?,?,?)");
        }
    }


    @Override
    public void invoke(T value, Context context) throws Exception {
        if (tClass.getName().equals(OutOfLimitSpeedInfo.class.getName())) {
            OutOfLimitSpeedInfo info = (OutOfLimitSpeedInfo) value;
            pst.setString(1, info.car);
            pst.setString(2, info.monitorId);
            pst.setString(3, info.roadId);
            pst.setDouble(4, info.realSpeed);
            pst.setInt(5, info.limitSpeed);
            pst.setLong(6, info.actionTime);
            pst.executeUpdate();
        }
        if (tClass.getName().equals(AvgSpeedInfo.class.getName())) {
            AvgSpeedInfo info = (AvgSpeedInfo) value;
            pst.setLong(1, info.start);
            pst.setLong(2, info.end);
            pst.setString(3, info.monitorId);
            pst.setDouble(4, info.avgSpeed);
            pst.setInt(5, info.carCount);
            pst.executeUpdate();
        }
        if (tClass.getName().equals(RepetitionCarWarning.class.getName())) {
            RepetitionCarWarning info = (RepetitionCarWarning) value;
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
            //当前的0时0分0秒
            Date day = sdf.parse(sdf.format(new Date(info.getActionTime())));
            //在一天中，每个车牌只写入一次
            PreparedStatement selectPst = conn.prepareStatement("select count(1) from t_violation_list where car=? and create_time between ? and ?");
            selectPst.setString(1, info.car);
            selectPst.setLong(2, day.getTime());
            selectPst.setLong(3, day.getTime() + (24 * 60 * 60 * 1000));
            ResultSet selectSet = selectPst.executeQuery();
            //当天没有该车牌号,插入
            if (selectSet.next() && selectSet.getInt(1) == 0) {
                pst.setString(1, info.car);
                pst.setString(2, info.msg);
                pst.setLong(3, info.actionTime);
                pst.executeUpdate();
            }
            selectSet.close();
            selectPst.close();
        }
    }

    @Override
    public void close() {
        try {
            pst.close();
            conn.close();
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }


}
